package cn._51doit.flink.day08;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 侧流数据
 * 本质上，就是将一个流中的数据打上标签，未打标签的数据流叫主流，
 * 打标签的数据流，叫侧流（非主流）
 *
 * 在Flink中，DataStream中的数据流，按照是否打标签，分为两类
 * 1.侧流，就是打上标签的数据流，想要获取，就必须根据标签才能获取到
 * 2.主流，就是未打标签的数据，直接操作该datastream就可以了
 *
 */
public class SideOutputDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2
        //3
        //abc
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);


        //奇数的标签
        OutputTag<Integer> oddTag = new OutputTag<Integer>("odd-tag") {};
        //偶数的标签
        OutputTag<Integer> evenTag = new OutputTag<Integer>("even-tag") {};
        //字符串类型标签
        OutputTag<String> strTag = new OutputTag<String>("str-tag") {};

        SingleOutputStreamOperator<String> mainStream = lines.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String input, Context ctx, Collector<String> out) throws Exception {

                try {
                    int num = Integer.parseInt(input);
                    if (num % 2 == 0) {
                        //偶数
                        ctx.output(evenTag, num);
                    } else {
                        //奇数
                        ctx.output(oddTag, num);
                    }
                } catch (NumberFormatException e) {
                    //字符串
                    ctx.output(strTag, input);
                }
                out.collect(input); //未打标签的数据流（就是对正常的数据流进行操作）
            }
        });

        //根据事先定义好的标签取出对应类型的数据
        DataStream<Integer> evenStream = mainStream.getSideOutput(evenTag);
        DataStream<String> strStream = mainStream.getSideOutput(strTag);

        evenStream.print("偶数流 ");
        strStream.print("字符串流 ");

        mainStream.print("主流 "); //打印的是未打标签的数据
        env.execute();

    }
}
